
package com.kweative.cluster;

import java.io.IOException;
import java.io.Serializable;
import java.net.InetAddress;
import java.net.MulticastSocket;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.logging.Level;
import java.util.logging.Logger;
import sun.tools.tree.ThisExpression;

/**
 *
 * @author henrik
 */
public class Cluster extends Object {
    private static Cluster instance;
    public static Cluster instance() {
        if (instance == null)
            instance = new Cluster();
        return instance;
    }

    

    

    private LinkedList<ClusterNode> nodes = new LinkedList<ClusterNode>();
    private ClusterNode currentNode = null;
    private Receiver receiver = new Receiver();
    private Transmitter transmitter = new Transmitter();
    private MulticastSocket socket;
    private int port = 5000;
    private String group = "225.4.5.6";
    private int ttl = 1;
    private ClusterNode master = null;
    private boolean ready = false;

    private HashMap<String,Serializable> datastore = new HashMap<String, Serializable>();

    public void connect() throws IOException, InterruptedException {
        if (isConnected()) throw new IOException("Cannot connect cluster twice!");
        socket = new MulticastSocket(port);
        socket.joinGroup(InetAddress.getByName(group));
        receiver.start();
        transmitter.start();
        int attempts = 0;
        int max = 20;
        while(!ready) {
            Thread.sleep(500);
            attempts++;
            if (attempts > max) break;
        }
        ready = true; //Ready or not... 
        if (nodes().size() == 1) {
            System.out.println(node() + " is lonely");
            setMaster(node());
        }
        Runtime.getRuntime().addShutdownHook(new Thread() {
            @Override
            public void run() {
                Cluster.instance().disconnect();
            }
        });
        System.out.println(node() + " is started");
    }
    public void disconnect() {
        transmit(new Action(Action.ActionType.LEAVE, null, null),false);
    }

    public void setDatastore(HashMap<String, Serializable> datastore) {
        this.datastore = datastore;
    }

    public ClusterNode getCurrentNode() {
        return currentNode;
    }

    public ClusterNode getMaster() {
        return master;
    }

    public void setMaster(ClusterNode master) {
        this.master = master;
    }

    public void setCurrentNode(ClusterNode currentNode) {
        this.currentNode = currentNode;
    }
    

    public LinkedList<ClusterNode> getNodes() {
        return nodes;
    }

    public void setNodes(LinkedList<ClusterNode> nodes) {
        this.nodes = nodes;
    }
    
    
    public boolean isConnected() {
        return (socket != null && socket.isBound());
    }

    public HashMap<String, Serializable> getDatastore() {
        return datastore;
    }
    
    public Receiver getReceiver() {
        return receiver;
    }

    public MulticastSocket getSocket() {
        return socket;
    }

    public Transmitter getTransmitter() {
        return transmitter;
    }
    public String getGroup() {
        return group;
    }

    public void setGroup(String group) {
        this.group = group;
    }

    public int getPort() {
        return port;
    }

    public void setPort(int port) {
        this.port = port;
    }

    public int getTtl() {
        return ttl;
    }

    public void setTtl(int ttl) {
        this.ttl = ttl;
    }

    public boolean isReady() {
        return ready;
    }

    public void setReady(boolean ready) {
        this.ready = ready;
    }
    

    public static void transmit(Action action) {
        instance().getTransmitter().addToQueue(action);
    }
    public static void transmit(Action action,boolean async) {
        if (async)
            instance().getTransmitter().addToQueue(action);
        else
            instance().getTransmitter().transmit(action);
    }
    public static Serializable get(String key) {
        if (instance().getDatastore().containsKey(key)) {
            return instance().getDatastore().get(key);
        }
        return null;
    }
    public static <R extends Serializable> R get(String key,R object) {
        if (instance().getDatastore().containsKey(key)) {
            return (R) instance().getDatastore().get(key);
        }
        store(key, object);
        return object;
    }
    public static void store(String key,Serializable object) {
        if (instance().getDatastore().containsKey(key)) {
            transmit(new Action(Action.ActionType.UPDATE,object, key));
        } else {
            transmit(new Action(Action.ActionType.INSERT,object, key));
        }
    }
    public static void remove(String key) {
        transmit(new Action(Action.ActionType.DELETE,null,key));
    }
    public static ClusterNode node() {
        return instance().getCurrentNode();
    }
    public static boolean isMaster() {
        return master() != null && master().equals(node());
    }
    public static ClusterNode master() {
        return instance().getMaster();
    }
    public static LinkedList<ClusterNode> nodes() {
        return instance().getNodes();
    }
    public static HashMap<String, Serializable> datastore() {
        return instance().getDatastore();
    }


}
